package com.hkbigdata.processapi;

import com.hkbigdata.bean.WaterSensor;
import org.apache.flink.api.common.functions.IterationRuntimeContext;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author liuanbo
 * @creat 2024-05-20-14:48
 * @see 2194550857@qq.com
 */
public class Flink03_ProcessJoinFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator = env.socketTextStream("hadoop102", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],
                                Long.valueOf(split[1]),
                                Integer.valueOf(split[2])
                        );
                    }
                });

        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator2 = env.socketTextStream("hadoop102", 7777)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],
                                Long.valueOf(split[1]),
                                Integer.valueOf(split[2])
                        );
                    }
                });


        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator1 = env.socketTextStream("hadoop102", 8888)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],
                                Long.valueOf(split[1]),
                                Integer.valueOf(split[2])
                        );
                    }
                });
        //实现类继承抽象类，需要重写里面的所有抽象方法
        //如果实现类实现接口，需要重写所有的方法，无论是否是抽象
        DataStream<WaterSensor> apply = waterSensorSingleOutputStreamOperator.keyBy(WaterSensor::getId).join(waterSensorSingleOutputStreamOperator1.keyBy(WaterSensor::getId))
                .where(WaterSensor::getId)
                .equalTo(WaterSensor::getId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .apply(new JoinFunction<WaterSensor, WaterSensor, WaterSensor>() {
                    @Override
                    public WaterSensor join(WaterSensor first, WaterSensor second) throws Exception {
                        return new WaterSensor(first.getId(), first.getTs(), first.getVc() + second.getVc());
                    }
                });

/*        apply.join(waterSensorSingleOutputStreamOperator2).where()
                .equalTo().window()
                .*/


        apply.addSink(new RichSinkFunction<WaterSensor>() {
            private Connection connection;
            private PreparedStatement preparedStatement;

            @Override
            public void open(Configuration parameters) throws Exception {
                connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test?useSSL=false", "root", "123456");
                preparedStatement = connection.prepareStatement("insert into  sensor values (?,?,?)");
            }

            @Override
            public void close() throws Exception {
                connection.close();
                preparedStatement.close();
            }

            @Override
            public void invoke(WaterSensor value, Context context) throws Exception {
                preparedStatement.setString(1, value.getId());
                preparedStatement.setLong(2, value.getTs());
                preparedStatement.setDouble(3, value.getVc());
                preparedStatement.execute();
            }
        });


        env.execute();
    }
}
